package org.kfive.nova.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.util.GenericOptionsParser;

import java.io.IOException;

/**
 * Created by admin on 2017-05-14.
 */
public class HbaseDriver {
    public static void main(String[] args) throws IOException {
        String table = "wordcount";
        Configuration conf = HBaseConfiguration.create();
        HBaseAdmin admin = new HBaseAdmin(conf);
        if(admin.tableExists(table)){
            System.out.println("table exists ! recreating!!!");
            admin.disableTable(table);
            admin.deleteTable(table);
        }
        HTableDescriptor htd = new HTableDescriptor(table);
        HColumnDescriptor hcd = new HColumnDescriptor("content");
        htd.addFamily(hcd);
        admin.createTable(htd);
        String[] otherArgs = new GenericOptionsParser(conf,args).getRemainingArgs();
        if(otherArgs.length!=1){
            System.out.println("Usage:wordcount <int> <out>");
            System.exit(2);
        }
        Job job = new Job(conf,"word count");
        job.setJarByClass(HbaseDriver.class);
//        job.setMapperClass(HbaseReaderMapper.class);
//        job.setReducerClass(HbaseWriterReducer.class);
        Scan scan = new Scan(otherArgs[0].getBytes());

        TableMapReduceUtil.initTableMapperJob(table,scan,HbaseReaderMapper.class,Text.class,Text.class,job);
        TableMapReduceUtil.initTableReducerJob(table,HbaseWriterReducer.class,job);
    }
}
